package com.lazy.lib.common;

import com.lazy.lib.lang.BoundedQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;

/**
 * 数据消费类, 处理队列的通用接口类
 * 
 * @author luodan
 *
 * @param <T>
 */
public abstract class DataConsumer<T> extends Thread {
    protected static final Logger logger = LoggerFactory.getLogger(DataConsumer.class);

    protected ExecutorService executor = null;

    protected BoundedQueue<T> queue;

    protected int batchSize = 1;

    protected long maxDelayTime = 500; // 处理数据最大延迟, 毫秒

    public DataConsumer(ExecutorService executor, BoundedQueue<T> queue){
        this.executor = executor;
        this.queue = queue;
    }
    
    /**
     * 设计这个方法的是, 有数据就运行, 没有数据就休眠
     */
    @Override
    public void run(){
        while (true) {
            try{
                if (batchSize <= 1) {
                    T item = queue.takeDataQueue();
                    processOne(item);
                } else {
                    List<T> list = new ArrayList<T>();
                    int count = queue.drainTo(list, batchSize);
                    if (count == 0) {
                        T item = queue.takeDataQueue();
                        processOne(item);
                        // TODO: 未来如果发现队列消费的太快, 达不到批量操作的要求, 可以在这个地方休息一小会
                        sleepMillis(maxDelayTime);
                    } else {
                        processList(list);
                    }
                }
            }catch(Exception e){
                logger.error("saver submit task with error.", e);
            }
        }
    }
    
    /**
     * 处理队列中的单个数据
     * @param data
     */
    protected abstract void processOne(T data);
    
    /**
     * 出列队列中的数据, 最大不超过batchsize
     * @param list
     */
    protected abstract void processList(List<T> list);
    
    protected void sleepMillis(long millis) {
        try{
            Thread.sleep(millis);
        }catch(InterruptedException e){
        }
    }
    
    public void setBatchSize(int batchSize){
        this.batchSize = batchSize;
    }
    
    public int getBatchSize(){
        return batchSize;
    }

    public void setMaxDelayTime(long maxDelayTime) {
        this.maxDelayTime = maxDelayTime;
    }

    public long getMaxDelayTime() {
        return maxDelayTime;
    }
}

